package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.BeanToJsonStrFunction;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.InteractionCourseReviewBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.DorisUtil;
import com.atguigu.edu.realtime.util.EnvAndCheckUtils;
import com.atguigu.edu.realtime.util.KafkaUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;


//互动域
public class DwsInteractionCourseReviewWindow {
    public static void main(String[] args) {
        // TODO 1 创建环境设置后端
        StreamExecutionEnvironment env = EnvAndCheckUtils.getEnvAndMarkCheck();

        // TODO 2 读取dwd对应主题dwd_interaction_review
        String topicName = "dwd_review_info";
        String groupId = "dwd_review_info_course_review_window";
        DataStreamSource<String> interactionReviewSource = env.fromSource(KafkaUtils.getKafkaSource(topicName, groupId), WatermarkStrategy.noWatermarks(), "interaction_review_source");
        //interactionReviewSource.print(">");

        // TODO 3 转换结构补充度量值
        SingleOutputStreamOperator<InteractionCourseReviewBean> beanStream = interactionReviewSource.flatMap(
                new FlatMapFunction<String, InteractionCourseReviewBean>() {
                    @Override
                    public void flatMap(String value, Collector<InteractionCourseReviewBean> out) throws Exception {
                        try {
                            JSONObject jsonObject = JSON.parseObject(value);
                            String courseId = jsonObject.getString("course_id");
                            Long reviewStars = jsonObject.getLong("review_stars");
                            long ts = jsonObject.getLong("ts") * 1000;
                            out.collect(InteractionCourseReviewBean.builder()
                                    .courseId(courseId)
                                    .reviewTotalStars(reviewStars)
                                    .reviewUserCount(1L)
                                    .goodReviewUserCount(reviewStars == 5L ? 1L : 0L)
                                    .ts(ts)
                                    .build());
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
        });
        //beanStream.print(">>>>>>>");

        // TODO 4 添加水位线
        SingleOutputStreamOperator<InteractionCourseReviewBean> withWaterMarkStream =
                beanStream.assignTimestampsAndWatermarks(WatermarkStrategy
                        .<InteractionCourseReviewBean>forMonotonousTimestamps().withTimestampAssigner(
                                new SerializableTimestampAssigner<InteractionCourseReviewBean>() {
                                    @Override
                                    public long extractTimestamp(InteractionCourseReviewBean element, long recordTimestamp) {
                                        return element.getTs();
                                    }
                                }
                        )
                );
        //withWaterMarkStream.print("水位线：");

        // TODO 5 按照课程id分组
        KeyedStream<InteractionCourseReviewBean, String> keyedStream = withWaterMarkStream.keyBy(new KeySelector<InteractionCourseReviewBean, String>() {
            @Override
            public String getKey(InteractionCourseReviewBean value) throws Exception {
                return value.getCourseId();
            }
        });

        // TODO 6 开窗聚合
        SingleOutputStreamOperator<InteractionCourseReviewBean> reduceStream = keyedStream.window(TumblingEventTimeWindows.of(Time.seconds(3L)))
                .reduce(new ReduceFunction<InteractionCourseReviewBean>() {
                    @Override
                    public InteractionCourseReviewBean reduce(InteractionCourseReviewBean value1, InteractionCourseReviewBean value2) throws Exception {
                        value1.setReviewUserCount(value1.getReviewUserCount() + value2.getReviewUserCount());
                        value1.setGoodReviewUserCount(value1.getGoodReviewUserCount() + value2.getGoodReviewUserCount());
                        value1.setReviewTotalStars(value1.getReviewTotalStars() + value2.getReviewTotalStars());
                        return value1;
                    }
                }, new ProcessWindowFunction<InteractionCourseReviewBean, InteractionCourseReviewBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<InteractionCourseReviewBean> elements, Collector<InteractionCourseReviewBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        String date = DateFormatUtil.toDate(context.window().getStart());

                        for (InteractionCourseReviewBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setCur_date(date);
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                });
        //reduceStream.print("聚合：");

        // TODO 7 维度关联课程信息
        SingleOutputStreamOperator<InteractionCourseReviewBean> dimBeanStream =
                AsyncDataStream.unorderedWait(
                        reduceStream,
                        new DimAsyncFunction<InteractionCourseReviewBean>("dim_course_info"){
                                @Override
                                public void join(InteractionCourseReviewBean bean, JSONObject jsonObj){
                                    bean.setCourseName(jsonObj.getString("course_name"));
                                }

                                @Override
                                public String getKey(InteractionCourseReviewBean bean) {
                                    return bean.getCourseId();
                                }
                        },
                        5 * 60, TimeUnit.SECONDS);
        dimBeanStream.print("关联：");

        //导入doris
        dimBeanStream.map(
                new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_interaction_course_review_window"));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
